#include <unistd.h>

#include <rtc_base/logging.h>
#include <rtc_base/crc32.h>
#include <rtc_base/rtc_certificate_generator.h>
#include <yaml-cpp/yaml.h>

#include "rtc_server.h"
#include "rtc_worker.h"

namespace xrtc {

// 默认指定证书的有效时间    
const uint64_t k_year_in_ms = 365 * 24 * 3600 * 1000L;    

RtcServer::RtcServer() : m_el(new EventLoop(this)) {

}

RtcServer::~RtcServer() {
    if (m_el) {
        delete m_el;
        m_el = nullptr;
    }

    if (m_thread) {
        m_thread.reset();
    }

    for (auto worker : m_workers) {
        if (worker) {
            delete worker;
            worker = nullptr;
        }
    }

    m_workers.clear();
}

void rtcServerRecvNotify(EventLoop *, IOWatcher *, int fd, int, void* data) {
    int msg;
    if (read(fd, &msg, sizeof(int)) != sizeof(int)) {
        RTC_LOG(LS_WARNING) << "read from pipe error: " << strerror(errno) << ", errno: " << errno;
        return;
    }

    RtcServer* server = (RtcServer*)data;
    server->_processNotify(msg);
}

int RtcServer::init(const char* conf_file) {
    if (!conf_file) {
        RTC_LOG(LS_WARNING) << "conf_file is null";
        return -1;
    }

    try {
        YAML::Node config = YAML::LoadFile(conf_file);
        RTC_LOG(LS_INFO) << "rtc server options:\n" << config;
        m_options.m_worker_num = config["worker_num"].as<int>();
    } catch (const YAML::Exception& e) {
        RTC_LOG(LS_WARNING) << "rtc server load conf file error: " << e.msg;
        return -1;
    }

    // 创建管道
    int fds[2];
    if (pipe(fds)) {
        RTC_LOG(LS_WARNING) << "create pipe error: " << strerror(errno) << ", errno: " << errno;
        return -1;
    }
    
    m_notify_recv_fd = fds[0];
    m_notify_send_fd = fds[1];

    m_pipe_watcher = m_el->createIOEvent(rtcServerRecvNotify, this);
    m_el->startIOEvent(m_pipe_watcher, m_notify_recv_fd, EventLoop::READ);

    // 创建多个worker线程
    for (int i = 0; i < m_options.m_worker_num; i++) {
         if (_createWorker(i) != 0) {
            return -1;
        }
    }

    return 0;
}




bool RtcServer::start() {
    if (m_thread) {
        RTC_LOG(LS_WARNING) << "rtc server already start";
        return false;
    }

    m_thread = std::make_unique<std::thread>([=]() {
        RTC_LOG(LS_INFO) << "rtc server event loop start";
        m_el->start();
        RTC_LOG(LS_INFO) << "rtc server event loop stop";
    });

    return true;
}

void RtcServer::stop() {
    notify(QUIT);
}

void RtcServer::join() {
    if (m_thread && m_thread->joinable()) {
        m_thread->join();
    }
}


void RtcServer::_processNotify(int msg) {
    switch (msg) {
        case QUIT:
            _stop();
            break;
        case RTC_MSG:
            _processRtcMsg();
            break;
        default:
            RTC_LOG(LS_WARNING) << "unknown msg: " << msg;
            break;
    }
}

void RtcServer::_stop() {
    m_el->deleteIOEvent(m_pipe_watcher);
    m_el->stop();
    close(m_notify_recv_fd);
    close(m_notify_send_fd);

    for (auto worker : m_workers) {
        if (worker) {
            worker->stop();
            worker->join();
        }
    }
}

int RtcServer::_createWorker(int worker_id) {
    RTC_LOG(LS_INFO) << "rtc server create worker, worker_id: " << worker_id;
    RtcWorker* worker = new RtcWorker(worker_id, m_options);

    if (worker->init() != 0) {
        return -1;
    }

    if (!worker->start()) {
        return -1;
    }

    m_workers.push_back(worker);

    return 0;
}

// 通过stream_name获取分配的rtc_worker
RtcWorker* RtcServer::_getWorker(const std::string &stream_name) {
    if (m_workers.size() == 0 || m_workers.size() != (size_t)m_options.m_worker_num) {
        return nullptr;
    }

    // CRC32​​ 是一种常用的哈希算法，它能够将任意长度的输入数据映射为一个固定长度（32位，即4字节）的哈希值
    uint32_t num = rtc::ComputeCrc32(stream_name.c_str(), stream_name.size());
    size_t index = num % m_options.m_worker_num;
    return m_workers[index];
}

// 生成校验ssl证书
int RtcServer::_generate_and_check_certificate() {
    // 证书是否已经生成或未过期
    if (!m_certificate || m_certificate->HasExpired(time(nullptr) * 1000)) {
        // 生成证书的算法
        rtc::KeyParams key_params;
        RTC_LOG(LS_INFO) << "DTLS enbaled, key type: " << key_params.type();
        // 生成证书
        m_certificate = rtc::RTCCertificateGenerator::GenerateCertificate(key_params, k_year_in_ms);
        if (m_certificate) { 
            // 打印生成的证书信息
            rtc::RTCCertificatePEM pem = m_certificate->ToPEM();
            RTC_LOG(LS_INFO) << "rtc certificate:\n" << pem.certificate();
        }
    }

    if (!m_certificate) {
        RTC_LOG(LS_WARNING) << "get certificate error";
        return -1;
    }

    return 0;
}

int RtcServer::notify(int msg) {
    int written = write(m_notify_send_fd, &msg, sizeof(int));
    return written == sizeof(int) ? 0 : -1;
}

// 向消息队列中发送数据
int RtcServer::sendRtcMsg(std::shared_ptr<RtcMsg> msg) {
    // 将消息投递到rtc_worker的队列中
    pushMsg(msg);
    return notify(RTC_MSG);
}

void RtcServer::pushMsg(std::shared_ptr<RtcMsg> msg) {
    std::lock_guard<std::mutex> lock(m_que_msg_mutex);
    m_que_msg.push(msg);
}

std::shared_ptr<RtcMsg> RtcServer::popMsg() {
    std::lock_guard<std::mutex> lock(m_que_msg_mutex);  
    if (m_que_msg.empty()) { 
        return nullptr;
    }

    std::shared_ptr<RtcMsg> msg = m_que_msg.front();
    m_que_msg.pop();

    return msg;
}

// 分发流消息给rtc_worker
void RtcServer::_processRtcMsg() {
    std::shared_ptr<RtcMsg> msg = popMsg();
    if (!msg) {
        return;
    }

    // 检查证书是否过期
    if (_generate_and_check_certificate() != 0) {
        return;
    }

    // 给消息添加证书
    msg->m_certificate = m_certificate.get();

    // 将流分配给rtc_worker进行处理（通过将流的名字进行哈希，保证同一个流被分配到同一个rtc_worker中）
    RtcWorker* worker = _getWorker(msg->m_stream_name);
    if (worker) {
        worker->sendRtcMsg(msg);
    }
}

} // namespace xrtc